package com.gaofei.springdemokafka.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 * kafka消费者
 */
@Component
public class KafkaConsumer {

    @Autowired
    private RedisTemplate<String ,String> redisTemplate;

    @KafkaListener(topics = {"2006A"},containerFactory="ackContainerFactory" ,id="ack")
    public void msg(ConsumerRecord<String, String> msgObject, Acknowledgment ack){
        String brokerKey = msgObject.key();
        //setNx  set if not exist
        //如果返回true,就代表着,首次存,能存进去  false:第二次存,存不进去了
        Boolean b = redisTemplate.opsForValue().setIfAbsent(brokerKey, "sss");

        //TODO
        if(!b){
            System.out.println("重复消费");//错误的思路
        }else{
//            继续消费
            System.out.println("消息是:"+msgObject.value());//已经收到消息了
            //开始修改数据的代码
            System.out.println("修改数据库");
        }

        System.out.println("消息的唯一标识:"+msgObject.key());
        //手工提交,让broker移动offsets-->让指针偏移,这行代码一定要放在消费者代码执行完之后
        ack.acknowledge();

    }
}
